/**
 * Copyright 2020 Inc.
 **/
package com.myz.netty.study.websocket;

import com.myz.netty.study.ChannelSupervise;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.handler.codec.http.*;
import io.netty.handler.codec.http.websocketx.*;
import io.netty.util.CharsetUtil;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.time.LocalDateTime;

import static io.netty.handler.codec.http.HttpUtil.isKeepAlive;

/**
 * TextWebSocketFrame 文本帧
 *
 * @author maoyz0621 on 20-2-27
 * @version v1.0
 */
public class WebsocketServerHandler extends SimpleChannelInboundHandler<Object> {

    private static final Logger LOGGER = LoggerFactory.getLogger(WebsocketServerHandler.class);

    private WebSocketServerHandshaker handshaker;

    @Override
    protected void channelRead0(ChannelHandlerContext ctx, Object msg) throws Exception {
        if (msg instanceof FullHttpRequest) {
            //以http请求形式接入，但是走的是websocket
            handleHttpRequest(ctx, (FullHttpRequest) msg);
        } else if (msg instanceof WebSocketFrame) {
            //处理websocket客户端的消息
            handlerWebSocketFrame(ctx, (WebSocketFrame) msg);
        } else {
            return;
        }
    }


    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        String longText = ctx.channel().id().asLongText();
        System.out.println(longText);
        String shortText = ctx.channel().id().asShortText();
        System.out.println(shortText);

        ChannelSupervise.sendToAll(new TextWebSocketFrame("用户[" + shortText + "]　上线了"));
        ChannelSupervise.addChannel(ctx.channel());
    }

    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        LOGGER.debug("================ WebsocketServerHandler handlerRemoved() ===============");

        Channel channel = ctx.channel();
        String shortText = ctx.channel().id().asShortText();

        ChannelSupervise.sendToAll(new TextWebSocketFrame("用户[" + shortText + "]　断开连接"));
        ChannelSupervise.removeChannel(channel);
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ctx.channel().close();
    }

    private void handlerWebSocketFrame(ChannelHandlerContext ctx, WebSocketFrame frame) {
        // 判断是否关闭链路的指令
        if (frame instanceof CloseWebSocketFrame) {
            handshaker.close(ctx.channel(), (CloseWebSocketFrame) frame.retain());
            return;
        }

        // 判断是否ping消息
        if (frame instanceof PingWebSocketFrame) {
            ctx.channel().write(new PongWebSocketFrame(frame.content().retain()));
            return;
        }

        // 暂只支持文本消息，不支持二进制消息
        if (!(frame instanceof TextWebSocketFrame)) {
            LOGGER.debug("本例程仅支持文本消息，不支持二进制消息");
            throw new UnsupportedOperationException(String.format("%s frame types not supported", frame.getClass().getName()));
        }

        // 返回应答消息
        String msg = ((TextWebSocketFrame) frame).text();
        LOGGER.debug("服务端收到：" + msg);
        TextWebSocketFrame tws = new TextWebSocketFrame(ctx.channel().id().asShortText() + " => " + LocalDateTime.now() + " say:" + msg);
        // 群发
        ChannelSupervise.sendToAll(tws);
    }

    /**
     * 唯一的一次http请求，用于创建websocket
     *
     * @param ctx
     * @param req
     */
    private void handleHttpRequest(ChannelHandlerContext ctx, FullHttpRequest req) {
        //要求Upgrade为websocket，过滤掉get/Post
        if (!req.decoderResult().isSuccess() || (!"websocket".equals(req.headers().get("Upgrade")))) {
            // 若不是websocket方式，则创建BAD_REQUEST的req，返回给客户端
            sendHttpResponse(ctx, req, new DefaultFullHttpResponse(HttpVersion.HTTP_1_1, HttpResponseStatus.BAD_REQUEST));
            return;
        }

        WebSocketServerHandshakerFactory wsFactory =
                new WebSocketServerHandshakerFactory("ws://127.0.0.1:8880/talk", null, false);
        handshaker = wsFactory.newHandshaker(req);
        if (handshaker == null) {
            WebSocketServerHandshakerFactory.sendUnsupportedVersionResponse(ctx.channel());
        } else {
            handshaker.handshake(ctx.channel(), req);
        }
    }

    /**
     * 拒绝不合法的请求，并返回错误信息
     */
    private static void sendHttpResponse(ChannelHandlerContext ctx, FullHttpRequest req, FullHttpResponse res) {
        // 返回应答给客户端
        if (HttpResponseStatus.OK.code() != res.status().code()) {
            ByteBuf buf = Unpooled.copiedBuffer(res.status().toString(), CharsetUtil.UTF_8);
            res.content().writeBytes(buf);
            buf.release();
        }

        ChannelFuture channelFuture = ctx.channel().writeAndFlush(res);
        // 如果是非Keep-Alive，关闭连接
        if (!isKeepAlive(req) || HttpResponseStatus.OK.code() != res.status().code()) {
            channelFuture.addListener(ChannelFutureListener.CLOSE);
        }
    }
}
